package com.csw.flink.source

import org.apache.flink.streaming.api.scala._

object Demo01Source {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)
    /**
      * 1、基于本地集合的构建DataStream
      *
      * 集合的数据是固定的，属于有界流
      */

    //返回一个集合
    val value: DataStream[List[Int]] = env.fromElements(List(1, 2, 3, 4, 5))
    //会自动拆开成一个一个的
    val listDS: DataStream[Int] = env.fromCollection(List(1, 2, 3, 4, 5, 6))

    //    value.print()
    //    listDS.print()

    /**
      * 2、基于文件构建DS
      *
      * flink 是事件驱动的应用程序，每一条数据都会处理，无论是有界流还是无界流都是一样的
      */

    val lineDS: DataStream[String] = env.readTextFile("flink/data/words.txt")

    val countDS: DataStream[(String, Int)] = lineDS.flatMap(_.split(",")).map((_,1)).keyBy(_._1).sum(1)

//    countDS.print()


    /**
      * 3、基于socket  构建DS
      */

    val socketDS: DataStream[String] = env.socketTextStream("master",8888)

    socketDS.print()

    /**
      * socket可以在linux中实现，安装后通过 nc -lk  8888(指定的端口号)进入
      */

    //启动flink
    env.execute()
  }
}
